package org.yuanzheng.transformation

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.yuanzheng.source.{CustomSource, StationLog}

/**
 * @author yuanzheng
 * @date 2020/6/17-13:18
 */
object TestTransformat {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    val stream: DataStream[StationLog] = streamEnv.addSource(new CustomSource)
    val result: DataStream[(String, Long)] = stream.filter(_.callType == "success").map(st => {
      (st.sid, st.duration)
    }).keyBy(0).reduce((st1, st2) => {
      var duration = st1._2 + st2._2
      (st1._1, duration)
    })
    result.print()
    streamEnv.execute()
  }
}
